package com.ruoyi.aimodel.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.JsonNode;
import com.github.yulichang.extension.mapping.config.DeepConfig;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
import com.google.gson.JsonSyntaxException;
import com.ruoyi.aimodel.domain.DeepSeekRequest;
import com.ruoyi.aimodel.domain.DeepSeekResponse;
import com.ruoyi.aimodel.service.DeepSeekService;
import com.ruoyi.common.redis.service.RedisService;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.logging.Logger;

@Service
public class DeepSeekServiceImpl implements DeepSeekService {

    private static final Logger logger = Logger.getLogger(DeepSeekServiceImpl.class.getName());

    private static final String USER = "user";
    private static final String SYSTEM = "system";
    private static final org.slf4j.Logger log = LoggerFactory.getLogger(DeepSeekServiceImpl.class);

    private final WebClient webClient;

    @Autowired
    private RedisService redisService;


    // 设置默认缓存时间为30分钟
    @Value("${youDeepseek.cache.expiration.minutes:30}")
    private int cacheExpirationMinutes;

    private final ObjectMapper objectMapper = new ObjectMapper();

    public DeepSeekServiceImpl(@Value("${deepseek.apiUrl}") String apiUrl,
                               @Value("${deepseek.apiKey}") String apiKey) {

        this.webClient = WebClient.builder()
                .baseUrl(apiUrl)
                .defaultHeader("Authorization", "Bearer " + apiKey)
                .build();
    }

    @Override
    public Flux<String> callDeepSeekStream(String userId, String userMessage) {
        return callDeepSeekAsyncStream(userId, userMessage);
    }

    @Override
    public String callDeepSeek(String userId, String userMessage) {
        Mono<String> stringMono = callDeepSeekAsync(userId, userMessage);
        log.info(new StringBuilder().append("接收返回：").append(stringMono).toString());
        return stringMono.block();
    }

    @Override
    public Flux<String> callDeepSeekStreamAipptOutline(String userId, String userMessage) {
        return callDeepSeekAsyncStreamAipptOutline(userId, userMessage);
    }

    @Override
    public Flux<String> callDeepSeekStreamAippt(String userId, String userMessage) {
        return callDeepSeekAsyncStreamAippt(userId, userMessage);
    }

    /**
     *  异步调用DeepSeek API
     *
     * @param userId
     * @param userMessage
     * @return {@link Mono }<{@link String }>
     */
    public Mono<String> callDeepSeekAsync(String userId, String userMessage) {
        String chatKey = getChatKey(userId);

        // 获取或初始化用户的对话历史
        Object object = redisService.getCacheMapValue(chatKey, String.valueOf(userId));
        List<DeepSeekRequest.Message> messages = Objects.isNull(object) ? new ArrayList<>() : (List<DeepSeekRequest.Message>) object;

        // 添加新的用户消息到对话历史
        messages.add(new DeepSeekRequest.Message(USER, userMessage));
        log.info("userId:"+userId);
        log.info("请求的消息体："+messages.toString());
        // 构建请求体
        DeepSeekRequest request = new DeepSeekRequest();
        request.setModel("deepseek-chat");
        request.setMessages(messages);

        // 发送异步请求
        return webClient.post()
                .bodyValue(request)
                .retrieve()
                .bodyToMono(DeepSeekResponse.class)
                .map(response -> {
                    if (response.getChoices() != null && !response.getChoices().isEmpty()) {
                        String content = response.getChoices().get(0).getMessage().getContent();
                        messages.add(new DeepSeekRequest.Message(SYSTEM, content));
//                        redisTemplate.opsForHash().put(chatKey, String.valueOf(userId), messages);
//                        redisTemplate.expire(chatKey, Duration.ofMinutes(cacheExpirationMinutes));
                        redisService.setCacheMapValue(chatKey, String.valueOf(userId), messages);
                        redisService.expire(chatKey, cacheExpirationMinutes);
                        log.info("content："+content);
                        return content;
                    } else {
                        throw new RuntimeException("Failed to call DeepSeek API: No choices in response");
                    }
                })
                .onErrorResume(e -> Mono.error(new RuntimeException("Failed to call DeepSeek API: " + e.getMessage())));
    }

    /**
     *  异步流式调用DeepSeek API
     *
     * @param userId
     * @param userMessage
     * @return {@link Flux }<{@link String }>
     */
    public Flux<String> callDeepSeekAsyncStream(String userId, String userMessage) {
        String chatKey = getChatKey(userId);

        // 获取或初始化用户的对话历史
        Object object = redisService.getCacheMapValue(chatKey, String.valueOf(userId));

        List<DeepSeekRequest.Message> messages = Objects.isNull(object) ? new ArrayList<>() : (List<DeepSeekRequest.Message>) object;

        // 添加新的用户消息到对话历史
        messages.add(new DeepSeekRequest.Message(USER, userMessage));
        log.info("请求的消息体："+messages.toString());
        // 构建请求体
        DeepSeekRequest request = new DeepSeekRequest();
        request.setModel("deepseek-chat");
        request.setMessages(messages);
        request.setStream(true);

        // 用于累积回答片段
        StringBuffer accumulatedContent = new StringBuffer();

        // 发送异步请求
        return webClient.post()
                .bodyValue(request)
                .retrieve()
                .bodyToFlux(String.class)
                .flatMap(chunk -> {
                    logger.info("接收部分: " + chunk);
                    if ("[DONE]".equals(chunk.trim())) {
                        return Flux.empty();
                    }
                    try {
                        JsonNode rootNode = objectMapper.readTree(chunk);
                        JsonNode choicesNode = rootNode.path("choices");
                        if (choicesNode.isArray() && !choicesNode.isEmpty()) {
                            JsonNode deltaNode = choicesNode.get(0).path("delta");
                            String content = deltaNode.path("content").asText("");
                            if (!content.isEmpty()) {
                                // 累积回答片段
                                accumulatedContent.append(content);
                                return Flux.just(content);
                            }
                        } else {
                            logger.warning("Choices 不能为空.");
                        }
                    } catch (Exception e) {
                        logger.severe("Failed to parse DeepSeek API response: " + e.getMessage());
                        return Flux.error(new RuntimeException("Failed to parse DeepSeek API response: " + e.getMessage()));
                    }
                    // 忽略非数据行
                    return Flux.empty();
                })
                .onErrorResume(e -> {
                    logger.severe("Failed to call DeepSeek API: " + e.getMessage());
                    return Flux.error(new RuntimeException("Failed to call DeepSeek API: " + e.getMessage()));
                })
                .doOnComplete(() -> {
                    // 确保在流式输出完成后保存完整的对话历史到 Redis
                    if (accumulatedContent.length() > 0) {
                        messages.add(new DeepSeekRequest.Message(SYSTEM, accumulatedContent.toString()));
                        redisService.setCacheMapValue(chatKey, String.valueOf(userId),messages);
                        redisService.expire(chatKey, cacheExpirationMinutes);

                    }
                });
    }



    /**
     *  异步流式调用DeepSeek API ppt
     *
     * @param userId
     * @param userMessage
     * @return {@link Flux }<{@link String }>
     */
    public Flux<String> callDeepSeekAsyncStreamAipptOutline(String userId, String userMessage) {

        List<DeepSeekRequest.Message> messages = new ArrayList<>() ;

        // 添加新的用户消息到对话历史
        messages.add(new DeepSeekRequest.Message(USER, userMessage));
        log.info("请求的消息体："+messages.toString());
        // 构建请求体
        DeepSeekRequest request = new DeepSeekRequest();
        request.setModel("deepseek-chat");
        request.setMessages(messages);
        request.setStream(true);

        // 用于累积回答片段
        StringBuffer accumulatedContent = new StringBuffer();
        StringBuffer sendContent = new StringBuffer();


        // 发送异步请求
        return webClient.post()
                .bodyValue(request)
                .retrieve()
                .bodyToFlux(String.class)
                .flatMap(chunk -> {
                    logger.info("接收部分: " + chunk);
                    if ("[DONE]".equals(chunk.trim())) {
                        return Flux.empty();
                    }
                    try {
                        JsonNode rootNode = objectMapper.readTree(chunk);
                        JsonNode choicesNode = rootNode.path("choices");
                        if (choicesNode.isArray() && !choicesNode.isEmpty()) {
                            JsonNode deltaNode = choicesNode.get(0).path("delta");
                            String content = deltaNode.path("content").asText("");
                            if (!content.isEmpty()) {
                                // 累积回答片段
                                if(content.contains("\n")){
                                    content=accumulatedContent.toString();
                                    accumulatedContent.setLength(0);
                                    return Flux.just(content);
                                }else {
                                    accumulatedContent.append(content);
                                }
                            }
                        } else {
                            logger.warning("Choices 不能为空.");
                        }
                    } catch (Exception e) {
                        logger.severe("Failed to parse DeepSeek API response: " + e.getMessage());
                        return Flux.error(new RuntimeException("Failed to parse DeepSeek API response: " + e.getMessage()));
                    }
                    // 忽略非数据行
                    return Flux.empty();
                })
                .onErrorResume(e -> {
                    logger.severe("Failed to call DeepSeek API: " + e.getMessage());
                    return Flux.error(new RuntimeException("Failed to call DeepSeek API: " + e.getMessage()));
                })
                .doOnComplete(() -> {
                });
    }

    /**
     *  异步流式调用DeepSeek API ppt
     *
     * @param userId
     * @param userMessage
     * @return {@link Flux }<{@link String }>
     */
    public Flux<String> callDeepSeekAsyncStreamAippt(String userId, String userMessage) {

        List<DeepSeekRequest.Message> messages = new ArrayList<>() ;

        // 添加新的用户消息到对话历史
        messages.add(new DeepSeekRequest.Message(USER, userMessage));
        log.info("请求的消息体："+messages.toString());
        // 构建请求体
        DeepSeekRequest request = new DeepSeekRequest();
        request.setModel("deepseek-chat");
        request.setMessages(messages);
        request.setStream(true);

        // 用于累积回答片段
        StringBuffer accumulatedContent = new StringBuffer();


        // 发送异步请求
        return webClient.post()
                .bodyValue(request)
                .retrieve()
                .bodyToFlux(String.class)
                .flatMap(chunk -> {
                    logger.info("接收部分: " + chunk);
                    if ("[DONE]".equals(chunk.trim())) {
                        return Flux.empty();
                    }
                    try {
                        JsonNode rootNode = objectMapper.readTree(chunk);
                        JsonNode choicesNode = rootNode.path("choices");
                        if (choicesNode.isArray() && !choicesNode.isEmpty()) {
                            JsonNode deltaNode = choicesNode.get(0).path("delta");
                            String content = deltaNode.path("content").asText("");
                            if (!content.isEmpty()) {
                                // 累积回答片段
                                accumulatedContent.append(content);
                                log.info("累积回答片段: " + accumulatedContent);
                                if((accumulatedContent.indexOf("{") != -1)&&isValidJsonWithGson(accumulatedContent.toString())){
                                    content=accumulatedContent.toString();
                                    accumulatedContent.setLength(0);
                                    log.info("返回的json: " + content);
                                    return Flux.just(content);
                                }
                            }
                        } else {
                            logger.warning("Choices 不能为空.");
                        }
                    } catch (Exception e) {
                        logger.severe("Failed to parse DeepSeek API response: " + e.getMessage());
                        return Flux.error(new RuntimeException("Failed to parse DeepSeek API response: " + e.getMessage()));
                    }
                    // 忽略非数据行
                    return Flux.empty();
                })
                .onErrorResume(e -> {
                    logger.severe("Failed to call DeepSeek API: " + e.getMessage());
                    return Flux.error(new RuntimeException("Failed to call DeepSeek API: " + e.getMessage()));
                })
                .doOnComplete(() -> {
                });
    }

    private String getChatKey(String userId) {
        return CacheConstant.DEEP_SEEK_CHAT_KEY + userId;
    }

    private static class CacheConstant {
        public static final String DEEP_SEEK_CHAT_KEY = "deep_seek_chat_";
    }

    public static boolean isValidJsonWithGson(String json) {
        try {
            JsonElement jsonElement = JsonParser.parseString(json);
            return true;
        } catch (JsonSyntaxException e) {
            return false;
        }
    }
}
